#include "raft/apply.h"

#include "raft/context.h"
#include "raft/meta.h"
#include "raft/raft.h"
#include "raft/raft.pb.h"
#include "raft/raft_log.h"

namespace libraft {
Apply::Apply(zrpc::EventLoop* loop, const std::shared_ptr<Context>& context,
             const std::shared_ptr<rocksdb::DB>& apply_db,
             const std::shared_ptr<Meta>& meta_db,
             const std::shared_ptr<RaftLog>& raft_log, Raft* raft)
    : loop_(loop),
      context_(context),
      apply_db_(apply_db),
      meta_db_(meta_db),
      raft_log_(raft_log),
      raft_(raft) {
  SetApplyLogCallback(
      std::bind(&Apply::DefaultApplyLogCallback, this));
}

Apply::~Apply() {}

void Apply::SetApplyLogCallback(ApplyLogCallback&& callback) {
  apply_log_callback_ = std::move(callback);
}

void Apply::ScheduleApply() {
  loop_->RunInLoop(std::bind(&Apply::ApplyStateMachine, this));
}

rocksdb::Status Apply::DefaultApplyLogCallback() {
  uint64_t last_applied = context_->last_applied;
  uint64_t commit_index = context_->commit_index;

  if (last_applied >= commit_index) {
    return rocksdb::Status::OK();
  }

  raft::Entry log_entry;
  while (last_applied < commit_index) {
    last_applied++;
    if (!raft_log_->GetEntry(last_applied, &log_entry)) {
      return rocksdb::Status::Corruption("parsed failed");
    }
    return ApplyLog(log_entry);
  }
}

rocksdb::Status Apply::ApplyLog(const raft::Entry& entry) {
  rocksdb::Status ret;
  std::string val;
  raft::Lock lock;

  switch (entry.optype()) {
    case raft::Entry_OpType_kWrite: {
      ret = apply_db_->Put(rocksdb::WriteOptions(), entry.key(), entry.value());
      break;
    }

    case raft::Entry_OpType_kDelete: {
      ret = apply_db_->Delete(rocksdb::WriteOptions(), entry.key());
      break;
    }

    case raft::Entry_OpType_kRead: {
      ret = rocksdb::Status::OK();
      break;
    }

    case raft::Entry_OpType_kTryLock: {
      ret = apply_db_->Get(rocksdb::ReadOptions(), entry.key(), &val);
      if (ret.ok()) {
        assert(lock.ParseFromString(val));
        if (lock.lease_end() < rocksdb::Env::Default()->NowMicros()) {
          lock.set_holder(entry.holder());
          lock.set_lease_end(entry.lease_end());
          lock.SerializeToString(&val);
          ret = apply_db_->Put(rocksdb::WriteOptions(), entry.key(), val);
        } else {
          ret = rocksdb::Status::OK();
        }
      } else if (ret.IsNotFound()) {
        lock.set_holder(entry.holder());
        lock.set_lease_end(entry.lease_end());
        lock.SerializeToString(&val);
        ret = apply_db_->Put(rocksdb::WriteOptions(), entry.key(), val);
      } else {
      }
      break;
    }

    case raft::Entry_OpType_kUnLock: {
      ret = apply_db_->Get(rocksdb::ReadOptions(), entry.key(), &val);
      if (ret.ok()) {
        if (lock.holder() != entry.holder()) {
        } else if (lock.lease_end() < rocksdb::Env::Default()->NowMicros()) {
        } else {
          ret = apply_db_->Delete(rocksdb::WriteOptions(), entry.key());
        }
      } else if (ret.IsNotFound()) {
        ret = rocksdb::Status::OK();
      } else {
      }
      break;
    }

    case raft::Entry_OpType_kAddServer: {
      ret = MembershipChange(entry.server(), true);
      if (ret.ok()) {
        context_->members_.insert(entry.server());
        raft_->AddPeerThread(entry.server());
      }
      break;
    }

    case raft::Entry_OpType_kRemoveServer: {
      ret = MembershipChange(entry.server(), false);
      if (ret.ok()) {
        context_->members_.erase(entry.server());
        assert(raft_->RemovePeerThread(entry.server()));
      }
      break;
    }

    case raft::Entry_OpType_kGetAllServers: {
      ret = rocksdb::Status::OK();
      break;
    }
  }
  return ret;
}

rocksdb::Status Apply::MembershipChange(const std::string& ip_port, bool add) {
  std::string value;
  raft::Membership members;
  rocksdb::Status s =
      apply_db_->Get(rocksdb::ReadOptions(), Raft::kMemberConfigKey, &value);
  if (!s.ok()) {
    return s;
  }

  if (!members.ParseFromString(value)) {
    return rocksdb::Status::Corruption("parse failed");
  }

  int count = members.nodes_size();
  for (int i = 0; i < count; i++) {
    if (members.nodes(i) == ip_port) {
      if (add) {
        return rocksdb::Status::OK();
      }

      if (i != count - 1) {
        std::string* nptr = members.mutable_nodes(i);
        *nptr = members.nodes(count - 1);
      }
      members.mutable_nodes()->RemoveLast();
    }
  }

  if (add) {
    members.add_nodes(ip_port);
  }

  if (!members.SerializeToString(&value)) {
    return rocksdb::Status::Corruption("serialize failed");
  }
  return apply_db_->Put(rocksdb::WriteOptions(), Raft::kMemberConfigKey, value);
}

void Apply::ApplyStateMachine() {
  if (apply_log_callback_) {
    rocksdb::Status s = apply_log_callback_();
    if (!s.ok()) {
      LOG_WARN << s.ToString();
      return;
    }
  }

  std::unique_lock<std::mutex> lk(context_->apply_mutex_);
  context_->last_applied = context_->commit_index;
  context_->last_applied_term = context_->current_term;
  meta_db_->SetLastApplied(context_->last_applied);
  context_->apply_cond_.notify_all();
}
}  // namespace libraft